﻿using System;
using System.Collections;
using System.Collections.Generic;
using NRank.Algorithms.Strategies;
using NRank.Common;

namespace NRank.Algorithms
{
	/// <summary>
	/// Implementation of HashRankJoin algorithm
	/// </summary>
	/// <typeparam name="T1">Type of left items</typeparam>
	/// <typeparam name="T2">Type of right items</typeparam>
	/// <typeparam name="TField">Type of equality field member</typeparam>
	/// <typeparam name="T">Type of output items</typeparam>
	public class HashRankJoin<T1, T2, TField, T> : IEnumerable<IRanked<T>>, IDepthAwareContainer
	{
		/// <summary>
		/// Equality field selector for left source
		/// </summary>
		private readonly Func<T1, TField> equalityFieldSelector1;

		/// <summary>
		/// Equality field selector for right source
		/// </summary>
		private readonly Func<T2, TField> equalityFieldSelector2;

		/// <summary>
		/// Join function
		/// </summary>
		private readonly Func<T1, T2, T> joinFunc;

		/// <summary>
		/// Pulling strategy
		/// </summary>
		private readonly PullingStrategyTypes pullingStrategy;

		/// <summary>
		/// Aggregate function
		/// </summary>
		private readonly Func<double, double, double> aggregateFunc;

		/// <summary>
		/// Join condition
		/// </summary>
		private readonly Func<T1, T2, bool> joinCondition;

		/// <summary>
		/// Left source
		/// </summary>
		private readonly DepthAware<IRanked<T1>> leftSource;

		/// <summary>
		/// Right source
		/// </summary>
		private readonly DepthAware<IRanked<T2>> rightSource;

		public int LeftDepth
		{
			get { return leftSource.Depth; }
		}

		public int RightDepth
		{
			get { return rightSource.Depth; }
		}		
		
		
		/// <summary>
		/// Creates instance of hash rank join algorithm
		/// </summary>
		/// <param name="leftSource">Left source</param>
		/// <param name="rightSource">Right source</param>
		/// <param name="equalityFieldSelector1">Equality field selector for left items</param>
		/// <param name="equalityFieldSelector2">Equality field selector for right items</param>
		/// <param name="rankFunc1">Function ranking left items</param>
		/// <param name="rankFunc2">Function ranking right items</param>
		/// <param name="joinFunc">Join function</param>
		/// <param name="pullingStrategy">Pulling strategy</param>
		/// <param name="aggregateFunc">Aggregate function</param>
		/// <param name="joinCondition">Join condition</param>
		public HashRankJoin(
			IEnumerable<T1> leftSource,
			IEnumerable<T2> rightSource,
			Func<T1, TField> equalityFieldSelector1,
			Func<T2, TField> equalityFieldSelector2,
			Func<T1, double> rankFunc1,
			Func<T2, double> rankFunc2,
			Func<T1, T2, T> joinFunc,
			PullingStrategyTypes pullingStrategy = PullingStrategyTypes.Alternating,
			Func<double, double, double> aggregateFunc = null,
			Func<T1, T2, bool> joinCondition = null)
		{
			if (leftSource == null) throw new ArgumentNullException("leftSource");
			if (rightSource == null) throw new ArgumentNullException("rightSource");
			if (equalityFieldSelector1 == null) throw new ArgumentNullException("equalityFieldSelector1");
			if (equalityFieldSelector2 == null) throw new ArgumentNullException("equalityFieldSelector2");
			if (rankFunc1 == null) throw new ArgumentNullException("rankFunc1");
			if (rankFunc2 == null) throw new ArgumentNullException("rankFunc2");
			if (joinFunc == null) throw new ArgumentNullException("joinFunc");

			this.equalityFieldSelector1 = equalityFieldSelector1;
			this.equalityFieldSelector2 = equalityFieldSelector2;

			this.joinFunc = joinFunc;
			this.pullingStrategy = pullingStrategy;
			this.aggregateFunc = aggregateFunc ?? Defaults.DefaultAggregateFunction;
			this.joinCondition = joinCondition;

			this.leftSource = leftSource.AsRanked(rankFunc1).AsDepthAware();
			this.rightSource = rightSource.AsRanked(rankFunc2).AsDepthAware();

		}



		/// <summary>
		/// Current threshold
		/// </summary>
		public double Threshold { get; private set; }

		/// <summary>
		/// Algorithm implementation
		/// </summary>
		/// <returns>Rank-joined items</returns>
		private IEnumerable<IRanked<T>> GetItems()
		{
			Func<IRanked<T1>, IRanked<T2>, IRanked<T>> joinFuncWithRank = (i1, i2) => joinFunc(i1.Object, i2.Object).AsRanked(aggregateFunc(i1.Rank, i2.Rank));

			var joinedResults = new Heap<IRanked<T>>();

			bool leftHasValue = true;
			bool rightHasValue = true;

			var leftAdapter = new HashedSource<T1, TField>(this.leftSource, equalityFieldSelector1);
			var left = leftAdapter.GetEnumerator();

			var rightAdapter = new HashedSource<T2, TField>(this.rightSource, equalityFieldSelector2);
			var right = rightAdapter.GetEnumerator();

			var thresholdCounter = new ThresholdCounter(leftAdapter, rightAdapter, this.aggregateFunc);

			IPullingStrategy pullStrategy = Defaults.GetPullStrategy(thresholdCounter, this.pullingStrategy);

			while (true)
			{
				bool canPullLeft = pullStrategy.CanPullLeft;
				if (canPullLeft && (leftHasValue = left.MoveNext()))
				{
					// pull row from left
					var leftItem = left.Current;
					var joinFieldValue = equalityFieldSelector1(leftItem.Object);

					// update threshold
					Threshold = thresholdCounter.CountThreshold();

					// join with all seen items - they are sorted implicitly by descending order - this is
					// important because we assume it in the inner while-loop when yielding the items as soon 
					// as possible
					foreach (var rightSeenItem in rightAdapter[joinFieldValue])
					{
						if (joinCondition == null || joinCondition(leftItem.Object, rightSeenItem.Object))
						{
							var newItem = joinedResults.Add(joinFuncWithRank(leftItem, rightSeenItem));
							//joinFunc(leftItem.Object, rightSeenItem.Object).AsRanked(aggregateFunc(leftItem.Rank, rightSeenItem.Rank)));

							bool metNewItem = false;
							while (joinedResults.Count > 0 && joinedResults.Peek().Rank >= Threshold && !metNewItem)
							{
								metNewItem = ReferenceEquals(joinedResults.Peek(), newItem);
								yield return joinedResults.RemoveTop();
							}
						}
					}

					while (joinedResults.Count > 0 && joinedResults.Peek().Rank >= Threshold)
					{
						yield return joinedResults.RemoveTop();
					}
				}

				bool canPullRight = pullStrategy.CanPullRight;
				if (canPullRight && (rightHasValue = right.MoveNext()))
				{
					// pull row from right
					var rightItem = right.Current;
					var joinFieldValue = equalityFieldSelector2(rightItem.Object);

					Threshold = thresholdCounter.CountThreshold();

					// join with all seen items
					foreach (var leftSeenItem in leftAdapter[joinFieldValue])
					{
						if (joinCondition == null || joinCondition(leftSeenItem.Object, rightItem.Object))
						{
							var newItem = joinedResults.Add(joinFuncWithRank(leftSeenItem, rightItem));

							bool metNewItem = false;
							while (joinedResults.Count > 0 && joinedResults.Peek().Rank >= Threshold && !metNewItem)
							{
								metNewItem = ReferenceEquals(joinedResults.Peek(), newItem);
								yield return joinedResults.RemoveTop();
							}
						}
					}

					while (joinedResults.Count > 0 && joinedResults.Peek().Rank >= Threshold)
					{
						yield return joinedResults.RemoveTop();
					}
				}

				if ((!canPullLeft || !leftHasValue) && (!canPullRight || !rightHasValue)) break;
			}

			// Inputs are exhausted, just empty the 
			while (joinedResults.Count > 0)
			{
				yield return joinedResults.RemoveTop();
			}
		}

		/// <summary>
		/// Enumerator implementation
		/// </summary>
		/// <returns>Result enumerator</returns>
		public IEnumerator<IRanked<T>> GetEnumerator()
		{
			return GetItems().GetEnumerator();
		}

		/// <summary>
		/// Non-generic enumerator implementation
		/// </summary>
		/// <returns>Non-generic result enumerator</returns>
		IEnumerator IEnumerable.GetEnumerator()
		{
			return GetEnumerator();
		}
	}
}